package org.kou.hadoop.io;

import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import org.msgpack.MessagePack;
import org.msgpack.type.Value;

import java.io.IOException;

public class MessagePackWritable implements WritableComparable<MessagePackWritable>{

    private Value value;
    
    private MessagePack msgPack = new MessagePack();
    
    public Value get(){
    	return value; 
    }

    public void set(Value value){
	this.value = value;
    }
    
    public MessagePackWritable(Value value){
    	this.value = value;
    }
    
    public MessagePackWritable(){}

    @Override
    public void readFields(DataInput input)throws IOException{
    	int size = input.readInt();
    	byte[] data = new byte[size];
    	input.readFully(data);
    	value = msgPack.read(data);
    }

    public void write(DataOutput output)throws IOException{
	
    	byte[] data = msgPack.write(value);
    	int size = data.length;
    	output.writeInt(size);
    	output.write(data);
    }

    public int compareTo(MessagePackWritable other){
    	if(this == other){
    		return 0;
    	}
    	return this.value.asIntegerValue().getInt() - other.get().asIntegerValue().getInt();
    }
}